/* 
 * File:   JServerMaster.cpp
 * Author: johnneyshi
 * 
 * Created on 2013年1月27日, 下午8:16
 */

#include <signal.h>


#include "JServerMaster.h"
#include <arpa/inet.h>
#include <iostream>
#include <string>
#include <string.h>
#include <stdio.h>
#include "JException.h"
#include "JClient.h"
#include <fstream>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include <boost/foreach.hpp>
#include "JLog.h"
#include "JDefine.h"
#include <sys/epoll.h>
#include "JEpoll.h"
#include "JAdapter.h"

using namespace std;



//std::queue<int> adapterConnFdQueue;

pthread_mutex_t needDealFdsLock = PTHREAD_MUTEX_INITIALIZER;


JServerMaster::JServerMaster()
{
}

//JServerMaster::JServerMaster(const JServerMaster& orig) : JServer(orig)
//{
//}

JServerMaster::~JServerMaster() 
{
}

void JServerMaster::Init(std::string _configFileName)
{
    ReadConfig(_configFileName);
}

void JServerMaster::Start()
{
    for(int i = 0;i < masterWorkerAdapters.size(); ++i)
    {
        AddWorker(masterWorkerAdapters[i]);
    }
    JServer::Start();
}

void JServerMaster::AddWorker(MasterWorkerAdapter& adapter)
{
    JClient* pClient = adapter.pClient;
    JServerWorker* pWorker = adapter.pWorker;
    JLog::Log(JLogMessage,"Add worker, client IP(%s) port(%d), worker IP(%s) port(%d)\n",
            pClient->GetClientIP().c_str(),pClient->GetClientPort(),pWorker->GetIP().c_str(),pWorker->GetPort());
    int pipe1[2],pipe2[2];
    pid_t childPid;
    pipe(pipe1);
    pipe(pipe2);
    if((childPid = fork()) == 0)
    {
        close(pipe1[0]);
        close(pipe2[1]);
        //pipe1[1] write; pipe2[0] read
        MasterWorkerAdapterMain(pipe2[0],pipe1[1],adapter);
        
        while(true)
        {
            usleep(30000);
        }
        cout << "can't print" << endl;
        //char buff[200];
        //read(connFd,buff,200);
        //cout << buff << endl;
    }
    close(pipe1[1]);
    close(pipe2[0]);
    //pipe1[0] read; pipe2[1] write
    //this->readPipes.push_back(pipe1[0]);
    //this->writePipes.push_back(pipe2[1]);
    adapter.readPipe = pipe1[0];
    adapter.writePipe = pipe2[1];
}

void JServerMaster::ReadConfig(std::string _configFileName)
{
    using boost::property_tree::ptree;
    using namespace boost::property_tree::xml_parser;
    ptree pt;
    read_xml(_configFileName, pt);
    std::string masterIP = pt.get<std::string>("ServerMaster.IP");
    int masterPort = pt.get<int>("ServerMaster.Port");
    JServer::Init(masterIP,masterPort);
    
    BOOST_FOREACH(ptree::value_type &v,pt.get_child("ServerMaster.ServerWorkers"))
    {
        MasterWorkerAdapter adapter;
        std::string clientIP = v.second.get<std::string>("Client.IP","");
        int clientPort = v.second.get<int>("Client.Port",0);
        JClient* pClient = new JClient(clientIP,clientPort);
        //this->pClients.push_back(pClient);
        adapter.pClient = pClient;
        
        std::string ip = v.second.get<std::string>("Server.IP");
        int port = v.second.get<int>("Server.Port");
        int weight = v.second.get<int>("Weight");
        int portCount = v.second.get<int>("PortCount");
        JServerWorker* pWorker = new JServerWorker;
        pWorker->Init(ip,port,weight);
        //this->pWorkers.push_back(pWorker);
        //this->portCount.push_back(portCount);
        adapter.pWorker = pWorker;
        adapter.portCount = portCount;
        this->masterWorkerAdapters.push_back(adapter);
    }
}

//把接受到的消息，加上fd，然后转发给其中的一个worker
void JServerMaster::DealEpollFd(int fd)
{
//    int workIndex = AssignWork();
////    JBuffer* buff = GetAvailableBuffer();
////    buff->Put((char*)&fd,4);
//    char b[1024];///////////////////////////这里可以优化，如果不用b，直接把read到的数据放进buff中，会快
//    int n = 0;
//    while(true)
//    {
//        bzero(b,1024);
//        n = read(fd,b,1024);
//        if(n <= 0)
//        {
//            break;
//        }
//        write(this->writePipes[workIndex],b,n);
//    }
}

//计算公式：现在负载量=当前正在处理的工作/权重，选择现在负载量最小的进行分配
int JServerMaster::AssignWork()
{
    float minCapacity = numeric_limits<float>::max();
    int minIndex = -1;
    for(int i = 0;i < this->masterWorkerAdapters.size(); ++i)
    {
        float f = (float)(masterWorkerAdapters[i].pWorker->GetCurrentWorkCount()) / masterWorkerAdapters[i].pWorker->GetWeight();
        if(minCapacity > f)
        {
            minCapacity = f;
            minIndex = i;
        }
    }
    return minIndex;
}



/*这里需要一个机制，因为read函数是如果在缓冲中有数据的话，就会一直读；比如对端连续write了两次数据，
 * 而在read的时候，如果数组大的话，可以一次性读完这两次write的数据；read函数不会意识到有两次write,
 * 所以这里在接受到的数据前面，必须要有一个指示接下来接受到数据的大小。
 * 鉴于：
 * int最大：2147483647
 * long最大：9223372036854775807
 * 所以加一个4字节的大小就够了
 * 
 * 
 * 模式是Adapter和客户通信，先用JBuffer接受来自客户的所有数据，然后由adapter转发给worker
 * 好处：如果不这样，那么在客户和adapter通信时，就把数据传给worker，由于客户到adapter之间传输比较慢，
 * 而adapter到worker之间是内部网络，速度很快，这样就浪费了网速。这叫做异步传输。
 */
//void JServerMaster::AdapterTransmitMsg(int readPipe, int writePipe, int toMasterConnFd,int toWorkerConnFd)
//{
//    JBuffer* pBuff = GetAvailableBuffer();
//    int n = 0,needReadCount = 0;
//    JEpoll epoll;
//    epoll.Create(3);
//    epoll.AddFd(readPipe);
//    epoll.AddFd(toMasterConnFd);
//    epoll.AddFd(toWorkerConnFd);
//    std::vector<int> sockFds = epoll.Wait();
//    for(int i = 0;i < sockFds.size(); ++i)
//    {
//        switch(sockFds[i])
//        {
//            case readPipe:
//                break;
//             
//            //这是从客户端发过来的信息，操作：把接受到的信息缓存，缓存完毕，发给toWorkerConnFd
//            case toMasterConnFd:
//                //首先读取数据块的大小
//                unsigned int size = 0;
//                read(toMasterConnFd,(void*)&size,SIZEOF_UNSIGNED_INT);
//                pBuff->Put((char*)&size,SIZEOF_UNSIGNED_INT);
//                char buff[BUFF_SIZE];
//                bzero(buff,BUFF_SIZE);
//                
//                while(size > 0)
//                {
//                    if(size > BUFF_SIZE)
//                    {
//                        needReadCount = BUFF_SIZE;
//                    }
//                    else
//                    {
//                        needReadCount = size;
//                    }
//                    n = read(toMasterConnFd,buff,needReadCount);
//                    size -= n;
//                    pBuff->Put(buff,n);
//                    //这里不许要用下面语句，因为是根据n来写道pBuff中的
//                    //bzero(buff,BUFF_SIZE);
//                }
//                //接受完数据，向worker转发
//                while((n = pBuff->Get(buff,BUFF_SIZE)) > 0)
//                {
//                    write(toWorkerConnFd,buff,n);
//                }
//                break;
//            //这是worker端发来的信息，由于接受的速度比较快，接受到数据后，不缓存，直接发送给toMasterConnFd
//            case toWorkerConnFd:
//                
//                break;
//        }
//    }
//    
//    
//    
//    
//
//}

void JServerMaster::DealNewConnfd(int newFd)
{
    //设置描述符为nonblock,为了防止拒绝攻击.
    SetNonBlock(newFd);
    int workIndex = AssignWork();
    //把得到的描述符传给下标为workIndex的进程
    write(newFd,(void*)&workIndex,sizeof(int));
    close(newFd);
}

/*Adapters是用于从队列中取出needDealFds,然后用缓存，得到所有的数据后，然后从ClientAdapters中选取一个
 * ClientAdapter，把数据发送给worker，然后接受来自worker处理完的数据，不需要缓存，直接传送给客户端，最后断开连接
 * 
 * 
 * 这里还需要一种机制：如果客户端拒绝攻击，或者发送的数据没有到size大小，那么就循环，从队列中另外再取一个描述符，进行处理
 * 但是原来的描述符，也用epoll监听着，最后只有接受到size大小的数据，才发往worker
 */
void* AdapterThread(void* arg)
{
    AdapterThreadArg* pArg = (AdapterThreadArg*)arg;
    JAdapter adapter;
    adapter.Start(pArg);
//////////////////////////////////////////////////////////
}


void JServerMaster::CreateClientAdapter(AdapterThreadArg* pArg)
{
    JClient client(pArg->pAdapter->pClient->GetClientIP(),pArg->port);
    client.Connect(pArg->pAdapter->pWorker->GetIP(),pArg->pAdapter->pWorker->GetPort());
    int toWorkerConnFd = client.GetConnFd();
    pArg->pMaster->toWorkerFds.push_back(toWorkerConnFd);
    //char msg[200];
    //bzero(msg,200);
    JLog::Log(JLogMessage,"Connect to worker success. Worker IP:%s, worker port:%d\n",pArg->pAdapter->pWorker->GetIP().c_str(),pArg->pAdapter->pWorker->GetPort());
}


/*
 * Adapters是用于从队列中取出needDealFds,然后用缓存，得到所有的数据后，然后从ClientAdapters中选取一个
 * ClientAdapter，把数据发送给worker。
 * ClientAdapter的用于发送数据给worker和接受来自worker的数据。
 */
void JServerMaster::MasterWorkerAdapterMain(int readFd,int writeFd,MasterWorkerAdapter& adapter)
{
    //设置线程最高优先级
//    pthread_attr_t attr;
    struct sched_param param;
//    pthread_attr_init(&attr);
//    pthread_attr_setschedpolicy(&attr,SCHED_RR);
    param.__sched_priority = 99;
//    pthread_attr_setschedparam(&attr,param);
    sched_setscheduler(getpid(),SCHED_RR,&param);
    /////////
    pthread_t tid;
//    //错：多线程，创建ClientAdapters////////////////////////////////////这里不许要多线程，只需要循环创建portCount个client就行了
    for(int i = 0;i < adapter.portCount; ++i)
    {
        AdapterThreadArg arg;
        arg.pAdapter = &adapter;
        arg.pMaster = this;
        arg.childReadPipe = readFd;
        arg.childWritePipe = writeFd;
        arg.port = adapter.pClient->GetClientPort() + i;
        CreateClientAdapter(&arg);
    }
    //多线程，创建2倍的Adapters，暂时定Adapters是ClientAdapters的2倍，因为Adapters要比ClientAdapters多，才会有意义
    for(int i = 0;i < 2 * adapter.portCount; ++i)
    {
        AdapterThreadArg arg;
        arg.pAdapter = &adapter;
        arg.pMaster = this;
        arg.childReadPipe = readFd;
        arg.childWritePipe = writeFd;
        pthread_create(&tid,NULL,&AdapterThread,(void*)&arg);
    }
    
    //把接受到的要处理的connFd都放入一个队列中，各个线程从中取connFd;
    while(true)
    {
        int newFd = -1;//
        read(readFd,(void*)&newFd,sizeof(int));
        pthread_mutex_lock(&needDealFdsLock);////////////////////加锁，这里有疑问，可能获得不到锁，就会读不到队列.
        needDealFds.push(newFd);
        pthread_mutex_unlock(&needDealFdsLock);
    }
}

